
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.sink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * 将 Student 集合数据sink到MySQL数据库中
 */
public class SinkMySQL {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source 定义 Student 对象
        DataStream<Student> studentDS = env.fromElements(new Student(null, "name", 18));

        studentDS.addSink(new RichSinkFunction<Student>() {
            Connection conn;
            PreparedStatement ps;
            boolean flag = true;
            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化操作，添加连接MySQL
                conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?useSSL=false"
                        , "root", "123456");
                String sql="INSERT INTO t_student(`id`,`name`,`age`) values(null,?,?)";
                ps = conn.prepareStatement(sql);
            }

            @Override
            public void invoke(Student value, Context context) throws Exception {
                ps.setString(1,value.getName());
                ps.setInt(2,value.getAge());
                ps.executeUpdate();
            }

            @Override
            public void close() throws Exception {
                ps.close();
                conn.close();
            }


        });
        //3.Transformation 暂时不需要
        //4.Sink 实现自定义 MySQL sink
        //5.execute

        //创建 Student 类，包含3个字段 id name age
        //创建 MySQLSink 类继承 RichSinkFunction<Student>
        //实现 open invoke close 方法
        env.execute();
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class Student {
    private String id;
    private String name;
    private int age;
}
